package streaming.demo5

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}

object SparkFlumePoll {

  def myUpdateFun(inpuntData:Seq[Int], sumData:Option[Int]) : Option[Int] = {
    val result: Int = inpuntData.sum+sumData.getOrElse(0)
    Some(result)
  }

  def main(args: Array[String]): Unit = {
    /**
      * Create a input stream from a Flume source.
      *
      * @param ssc          StreamingContext object
      * @param hostname     Hostname of the slave machine to which the flume data will be sent
      * @param port         Port of the slave machine to which the flume data will be sent
      * @param storageLevel Storage level to use for storing the received objects
      *                     SparkFlumeEvent封装了event事件，每个event就是flume采集的数据
      */
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkFlumePoll").setMaster("local[4]")
    val sc: SparkContext = new SparkContext(sparkConf)
    val streamingContext: StreamingContext = new StreamingContext(sc, Seconds(5))
    //保存历史数据
    streamingContext.checkpoint("./checkflumepoll")
    val pollingStream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createPollingStream(streamingContext, "node03", 8888, StorageLevel.MEMORY_AND_DISK_2)
    //event 中的数据格式{"headers":xxxx,"body":xxxx} body是我们想要的数据
    //x.event获取event对象，x.event.getBody只获取body当中的bytebuffer的数据
    //x.event.getBody.array() 将bytebuffer的数据封装到一个数组里面去，将字节类型的数组转换成为string类型的字符串
    val map: DStream[String] = pollingStream.map(x => new String(x.event.getBody.array()))
    val map1: DStream[String] = map.flatMap(x => x.split(" "))
    val wordAndOne: DStream[(String, Int)] = map1.map((_, 1))
    //updatastatebykey针对的是key，value型的数据 别的数据类型不行 不断的累加历史数据
    val finalResult: DStream[(String, Int)] = wordAndOne.updateStateByKey(myUpdateFun)
    finalResult.print()
    map.print()
    streamingContext.start()
    streamingContext.awaitTermination()

  }
}
